package com.sky.test.kafka;

import net.minidev.json.JSONUtil;
import org.apache.kafka.clients.producer.*;

import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @author: 朱海鹏
 * @Package: com.sky.test.kafka
 * @Project: sky-take-out
 * @date: 2025/2/20 18:40
 * @description : kafka 生产者
 */
public class ProducerQuickStart {

    public static void main(String[] args) {
        // 1 设置kafka的配置信息
        Properties props = new Properties();
        // 1.1 设置kafka的连接地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
        // 1.1 key 和 value的序列化对象
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 2 创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        // ack配置，消息确认机制
        /**
         * acks = 0 生产者在成功写入消息之前不会等待任何来自服务器的响应，消息有丢失的风险，但是速度最快
         * acks = 1(默认值) 只要集群首领节点收到信息，生产者就会收到一个来自服务器的成功响应
         * asks = all 只有当所有参与赋值的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应
         */
        props.put(ProducerConfig.ACKS_CONFIG, 0);

        // 设置重试次数
        /**
         * 生产者从服务器收到的错误有可能是临时性错误，在这种情况下，retries参数的值决定了生产者可用重发消息的次数，如果达到这个次数
         * 生产者会放弃重试返回错误，默认情况下，生产者会在每次重试之间等待100ms
         */
        props.put(ProducerConfig.RETRIES_CONFIG, 10);

        // 设置消息压缩
        /**
         * snappy : 占用较少的CPU，却能提供较好的性能和相当可观的压缩比，如果看在网络带宽，建议采用
         * lz4:占用较少的CPU，压缩和解压速度较快，压缩比也很可观
         * gzip: 占用较多的CPU，但会提供更高的压缩比，网络带宽有限，可用使用这种算法
         * 使用压缩可用降低网络传输和存储开销，而往往是向Kafka发送消息的瓶颈所在
         */
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"gzip");
        // 3 发送消息
        /**
         * 第一个参数：topic 主题
         * 第二个key 消息密匙key
         * 第三个参数 value参数值
         */

        ProducerRecord<String, String> record = new ProducerRecord<>("topic1", "key-oo1", "朱海鹏");
        // 这边同步步发送
    /*    try {
            RecordMetadata recordMetadata = producer.send(record).get();
            System.out.println(recordMetadata.offset());
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }*/

        // 这边采用异步的方式进行发送消息
        try {
            producer.send(record, (recordMetadata, e) -> {
                if (e != null) {
                    System.out.println("Error: " + e.getMessage());
                }
                System.out.println(recordMetadata.offset());
            });
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        //4 关闭消息通道
        producer.close();
    }
}
